import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

/**
 * 任务二：商家周边活跃顾客数量统计
 * 统计每个商家与周边消费者的距离分布（按人数去重）
 * 输出格式：<Merchant_id> TAB <距离为x的消费者人数>
 */
public class Task2MerchantDistance {

    public static class DistanceMapper extends Mapper<Object, Text, Text, Text> {
        private Text merchantDistance = new Text();
        private Text userId = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(",", -1);
            
            if (fields.length < 7 || fields[0].equals("User_id")) {
                return; // 跳过表头
            }

            String user = fields[0];       // User_id
            String merchantId = fields[1]; // Merchant_id
            String distance = fields[4];   // Distance

            if (merchantId == null || merchantId.equals("null") || merchantId.isEmpty()) {
                return;
            }
            if (user == null || user.equals("null") || user.isEmpty()) {
                return;
            }

            // 距离为NULL的记录也统计
            if (distance == null || distance.equals("null") || distance.isEmpty()) {
                distance = "NULL";
            }

            // 输出：<商家ID_距离> -> <用户ID>
            merchantDistance.set(merchantId + "\t" + distance);
            userId.set(user);
            context.write(merchantDistance, userId);
        }
    }

    public static class DistanceReducer extends Reducer<Text, Text, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            // 使用 HashSet 对用户ID去重
            Set<String> uniqueUsers = new HashSet<>();
            for (Text val : values) {
                uniqueUsers.add(val.toString());
            }
            result.set(uniqueUsers.size());
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Task2 Merchant Distance Stats");
        job.setJarByClass(Task2MerchantDistance.class);
        job.setMapperClass(DistanceMapper.class);
        job.setReducerClass(DistanceReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
